﻿using System.Collections.Generic;
using System;
using System.Diagnostics;
using System.Linq;
using System.Linq.Expressions;
using System.Net;
using System.Text;
using System.Threading;
using GPC_Overlay;
using Microsoft.WindowsAzure;
using Microsoft.WindowsAzure.ServiceRuntime;
using Microsoft.WindowsAzure.StorageClient;
using Microsoft.WindowsAzure.Diagnostics;
using System.IO;
using cloudMPI;

namespace Master
{
    public class WorkerRole : RoleEntryPoint
    {
        private string connectionString = string.Empty;
        int numMessages = 0;
        private static List<Polygon> basePoly;
        private static List<Polygon> overlayPoly;
        private static ManualResetEvent[] eventsX;
        public static int currentOps = 0;
        public TimeSpan fileDownloadTime = new TimeSpan(0, 0, 0); /* Time to download files from cloud storage */
        public TimeSpan parseTime = new TimeSpan(0, 0, 0); /* time to download files and parse them into C# objects */
        public TimeSpan graphCreation = new TimeSpan(0, 0, 0);/* Time to create intersection graph */
        public TimeSpan totalTaskCreationTime = new TimeSpan(0, 0, 0); /* Time to convert each task to XML and writing it to Cloud storage */
        public TimeSpan taskUploadingTime = new TimeSpan(0, 0, 0); /* Time to write tasks to cloud storage */
        public TimeSpan taskDownloadingTime = new TimeSpan(0, 0, 0); /* Time to download tasks for processing from cloud */
        public TimeSpan processingTime = new TimeSpan(0, 0, 0); /* time to download a task, convert to polygons, and process using third party library */
        string _map1, _map2, _op;

        public override void Run()
        {
            // This is a sample worker implementation. Replace with your logic.
            connectionString = RoleEnvironment.GetConfigurationSettingValue("DataConnectionString");
            //GPC_Overlay.QueueHelper  QueueObject = new GPC_Overlay.QueueHelper(connectionString);
            //CloudQueueMessage msg = new CloudQueueMessage("");
            cMPI mpi = new cMPI(connectionString, "Master");
            cMPI_Message inputMsg = new cMPI_Message();
            string[] splitMsg;
            //string inputMsg = String.Empty;
            //while (inputMsg == String.Empty)
            //{
            //    if (QueueObject.PeekMessage(GPC_Overlay.DataConstants.INPUT_PARAS, out msg))
            //        inputMsg = msg.AsString;
            //    else
            //        Thread.Sleep(5000);
            //}
            while (!mpi.Recv(out inputMsg, "0", "input"))
            {
                //loop until you get the message
            }
            mpi.Init(0, RoleEnvironment.Roles["GISWorkerRole"].Instances.Count - 1, "overlay");
            try
            {
                splitMsg = inputMsg.data.ToString().Split('{');
                _map1 = splitMsg[0]; _map2 = splitMsg[1]; _op = splitMsg[2];

                string note = "Got the inputs and parsing starts at: " + DateTime.Now.ToString();
                System.Diagnostics.Trace.WriteLine(note, "Information");

                DateTime parsingStart = DateTime.Now;
                _parseFiles();
                parseTime = parseTime.Add(DateTime.Now.Subtract(parsingStart));
                note = "Parsing ends and task creation starts at: " + DateTime.Now.ToString();
                System.Diagnostics.Trace.WriteLine(note, "Information");


                DateTime taskCreationStart = DateTime.Now;
                _createGraph();
                graphCreation = graphCreation.Add(DateTime.Now.Subtract(taskCreationStart));
                note = "task creation ends and blob uploading starts at: " + DateTime.Now.ToString();
                System.Diagnostics.Trace.WriteLine(note, "Information");


                DateTime writingStart = DateTime.Now;
                _writeTasks();
                totalTaskCreationTime = totalTaskCreationTime.Add(DateTime.Now.Subtract(writingStart));
                note = "Blob uploading ends and overlay processing starts at " + DateTime.Now.ToString();
                System.Diagnostics.Trace.WriteLine(note, "Information");
            }
            catch(Exception e)
            {
                System.Diagnostics.Trace.WriteLine("Error in Master:" + e.Message, "Error");
            }

            while (true) ;
        }

        public override bool OnStart()
        {
            ServicePointManager.DefaultConnectionLimit = 48;

            // For information ons handling configuration changes
            // see the MSDN topic at http://go.microsoft.com/fwlink/?LinkId=166357.
            CloudStorageAccount.SetConfigurationSettingPublisher((configName, configSetter) =>
            {
                // Provide the configSetter with the initial value
                configSetter(RoleEnvironment.GetConfigurationSettingValue(configName));

                RoleEnvironment.Changed += (sender, arg) =>
                {
                    foreach (var change in arg.Changes)
                    {
                        var configChange = change as RoleEnvironmentConfigurationSettingChange;
                        if (configChange != null)
                        {
                            RoleEnvironment.GetConfigurationSettingValue(
                                configChange.ConfigurationSettingName);
                        }
                    }
                };
            });


            var diag = DiagnosticMonitor.GetDefaultInitialConfiguration();
            diag.Logs.ScheduledTransferLogLevelFilter = LogLevel.Verbose;
            diag.Logs.ScheduledTransferPeriod = TimeSpan.FromMinutes(1.0);
            diag.WindowsEventLog.DataSources.Add("Application!*");
            diag.WindowsEventLog.DataSources.Add("System!*");
            diag.WindowsEventLog.ScheduledTransferPeriod = TimeSpan.FromMinutes(1.0);
            diag.DiagnosticInfrastructureLogs.ScheduledTransferPeriod = TimeSpan.FromMinutes(1.0);
            DiagnosticMonitor.Start("Microsoft.WindowsAzure.Plugins.Diagnostics.ConnectionString", diag);

            return base.OnStart();


        }
        private void _parseFiles()
        {
            DateTime removeStart = DateTime.Now;
            Parser layer1 = null, layer2 = null;

            List<string> msgids = new List<string>();

            /* Create space in local hard disk of this instance */
            LocalResource localCache = RoleEnvironment.GetLocalResource("maps");
            string localCacheRootDirectory = localCache.RootPath;

            /* download source XML files */

            string mapFirst = Path.Combine(localCacheRootDirectory, "base");
            string mapSecond = Path.Combine(localCacheRootDirectory, "overlay");
            GPC_Overlay.BlobHelper blobObject = new GPC_Overlay.BlobHelper(connectionString);
            string note;
            note = "File Downloading starts at: " + DateTime.Now.ToString();
            System.Diagnostics.Trace.WriteLine(note, "Information");
            DateTime downloadtime = DateTime.Now;
            {
                blobObject.BlobStreamDownload(GPC_Overlay.DataConstants.MAP_CONTAINER_BLOB, _map1, mapFirst);
                blobObject.BlobStreamDownload(GPC_Overlay.DataConstants.MAP_CONTAINER_BLOB, _map2, mapSecond);
            }
            note = "File Downloading ends at: " + DateTime.Now.ToString();
            System.Diagnostics.Trace.WriteLine(note, "Information");
            //fileDownloadTime = fileDownloadTime.Add(DateTime.Now.Subtract(downloadtime));

            /* Parse GML files */
            layer1 = new Parser(mapFirst);
            layer2 = new Parser(mapSecond);
            numMessages = layer1.poly.Count;

            //GPC_Overlay.QueueHelper.PutMessageStatic(connectionString, GPC_Overlay.DataConstants.BLOB_COUNT, new CloudQueueMessage(numMessages.ToString()));
            /* Send the message to the master to notify it of the total message count */
            cMPI mpi = new cMPI(connectionString, "Master");
            mpi.Init(0, 0, "input");
            cMPI_Message message = new cMPI_Message();
            message.dest = "0";/* The destination is 0 for Master worker role */
            message.data = numMessages.ToString();
            mpi.Send(message, "input", true);
            basePoly = new List<Polygon>();
            overlayPoly = new List<Polygon>();
            basePoly = layer1.poly;
            overlayPoly = layer2.poly;
            layer1.poly = null;
            layer2.poly = null;
        }
        private void _createGraph()
        {
            /* Create the graph of relationships between web role and worker roles */
            ParallelSort.QuicksortParallel(overlayPoly, true);
            Parser.createRelationships(ref basePoly, ref overlayPoly);
            overlayPoly.Clear();
        }
        private void _writeTasks()
        {
            StringBuilder task = new StringBuilder();
            int concurrencyLimit = int.Parse(RoleEnvironment.GetConfigurationSettingValue("concurrencyLimit"));
            
            eventsX = new ManualResetEvent[concurrencyLimit];

            while (currentOps < concurrencyLimit)
            {
                eventsX[currentOps] = new ManualResetEvent(false);
                ThreadInfo para = new ThreadInfo();
                int from = 0, to = 0;
                int remainder = numMessages % concurrencyLimit;
                from = currentOps * ((int)Math.Floor((double)numMessages / concurrencyLimit)) + Math.Min(currentOps, remainder);
                to = (currentOps + 1) * ((int)Math.Floor((double)numMessages / concurrencyLimit)) + Math.Min(currentOps + 1, remainder) - 1;
                para.fromPoly = from;
                para.toPoly = to;
                para.threadId = currentOps;
                ThreadPool.QueueUserWorkItem(new WaitCallback(_uploadBlob), (object)para);
                Interlocked.Increment(ref currentOps);
            }
            WaitHandle.WaitAll(eventsX);
        }
        private static void _uploadBlob(object parameters)
        {
            /* Get MPI and message objects */
            string connectionString = RoleEnvironment.GetConfigurationSettingValue("DataConnectionString");
            cMPI mpi = new cMPI(connectionString, "GISWorkerRole");
            cMPI_Message blob = new cMPI_Message();
            /* Read threadinfo */
            ThreadInfo processingRange = (ThreadInfo)parameters;
            int from = processingRange.fromPoly;
            int to = processingRange.toPoly;
            int threadID = processingRange.threadId;
            int procCount = RoleEnvironment.Roles["GISWorkerRole"].Instances.Count;
            StringBuilder task = new StringBuilder();
            try
            {
                for (int i = from; i <= to; i++)
                {
                    task.Clear();
                    task.Append(Parser.PolygonToString(basePoly[i]));
                    foreach (Polygon overlayPolygon in basePoly[i].overlayPolygons)
                    {
                        string polygonasstring = Parser.PolygonToString(overlayPolygon);
                        if (polygonasstring.Length > 0)
                        {
                            task.Append("[" + polygonasstring);
                        }
                    }
                    //string id = Guid.NewGuid().ToString();
                    /* Let us send this message in a round robin fashion */
                    blob.data = task.ToString();
                    blob.dest = (i % procCount).ToString();
                    mpi.Send(blob, "overlay");
                    //GPC_Overlay.BlobHelper.PutBlobText(connectionString, GPC_Overlay.DataConstants.OUTPUT_CONTAINER_BLOB, id, task.ToString());
                    //GPC_Overlay.QueueHelper.PutMessageStatic(connectionString, GPC_Overlay.DataConstants.WORKER_QUEUE, new CloudQueueMessage(id));
                }
            }
            catch (Exception e)
            {
                System.Diagnostics.Trace.WriteLine("Error processing " + e.Message, "Information");
            }
            eventsX[threadID].Set();
        }
    }
}
